package com.shujia

import java.io
import scala.io.Source

object Code37FullJoin {
  def main(args: Array[String]): Unit = {
    val stuInfoList: List[(String, String, Int, String, String)] = Source
      .fromFile("scala_code/data/students_join.txt")
      .getLines()
      // 将读取到的每一行数据转换成List中的元素
      .toList
      .map(
        (line: String) => {
          val colum: Array[String] = line.split(",")
          // (id,name,age,gender,clazz)
          //  Tuple相比较Array 其元素的数据类型可以多样
          (colum(0), colum(1), colum(2).toInt, colum(3), colum(4))
        }
      )

    val scoreList: List[(String, String, String)] = Source
      .fromFile("scala_code/data/score_join.txt")
      .getLines()
      // 将读取到的每一行数据转换成List中的元素
      .toList
      .map(
        line => {
          val column: Array[String] = line.split(",")
          (column(0), column(1), column(2))
        }
      )


    // 全连接
    // 1.从 stuInfoList 中取出每个学生数据 再和成绩数据进行关联，关联方式选择使用 Left join
    // 2.要从 scoreList 中过滤出 stuInfoList 中没有的ID ，并返回和  Left join 结果类型一致的数据
    // 3.对1和2的数据进行拼接


    val groupByRes: Map[String, List[(String, String, String)]] = scoreList
      .groupBy(_._1)


    // left join
    val leftJoinRes: List[(String, String, Int, String, String, io.Serializable, io.Serializable)] = stuInfoList.flatMap(
      info => {
        if (groupByRes.contains(info._1)) {
          groupByRes(info._1).map(
            oneIDAllScore => {
              (info._1, info._2, info._3, info._4, info._5, oneIDAllScore._2, oneIDAllScore._3)
            }
          )
        } else {
          List((info._1, info._2, info._3, info._4, info._5, None, None))
        }
      }
    )


    // 由于leftJoinRes中有之前关联的ID，那么需要遍历groupByRes中所有的ID并和  leftJoinRes中ID判断是否存在

    val letJoinID: List[String] = leftJoinRes.map(
      _._1
    ).distinct


    groupByRes
      .filter(
        oneIDAllScore =>
          // 如果包含ID 说明关联过了 不需要,不包含则取出数据
          !letJoinID.contains(oneIDAllScore._1)
      )
      .flatMap(
        oneIDAllScore => {

          oneIDAllScore._2.map(
            oneScore =>
              (oneScore._1, None, None, None, None, oneScore._2, oneScore._3)
          )
        }
      )
      .toList
      .union(leftJoinRes)
      .foreach(println)


  }
}
